package com.skyline.courier.mq.provider.gearman;

import org.gearman.client.GearmanClient;
import org.gearman.client.GearmanClientImpl;
import org.gearman.common.GearmanNIOJobServerConnectionFactory;

import com.skyline.courier.mq.ExchangeType;
import com.skyline.courier.mq.MessageSendClient;
import com.skyline.courier.mq.MessageSendClientPool;

/**
 * Gearman消息发送（作业提交）客户端池
 * 
 * @author wuqh
 * 
 */
public class GearmanMessageSendClientPool implements MessageSendClientPool {
	private GearmanClient gearmanClient = new GearmanClientImpl();
	private String hosts;

	private GearmanAddress[] knownHosts;
	private boolean isClentInited = false;

	@Override
	public MessageSendClient getClient() {
		initGearmanClient();
		GearmanMessageSendClient client = new GearmanMessageSendClient();
		client.setClient(gearmanClient);
		return client;
	}

	@Override
	public void setExchangeName(String exchangeName) {
		throw new UnsupportedOperationException("Gearman的消息机制不支持exchangeName");
	}

	@Override
	public void setExchangeType(ExchangeType type) {
		throw new UnsupportedOperationException("Gearman的消息机制不支持ExchangeType");
	}

	@Override
	public void setHost(String hosts) {
		this.hosts = hosts;
	}

	@Override
	public void destroy() {
		if (this.gearmanClient != null && !this.gearmanClient.isShutdown()) {
			this.gearmanClient.shutdown();
		}
	}

	/**
	 * 初始化Gearman作业提交客户端
	 * 
	 */
	public void initGearmanClient() {
		if (isClentInited) {
			return;
		}
		GearmanNIOJobServerConnectionFactory connectionFactory = new GearmanNIOJobServerConnectionFactory();
		knownHosts = GearmanAddress.parseAddresses(hosts);
		for (GearmanAddress host : knownHosts) {
			gearmanClient.addJobServer(connectionFactory.createConnection(host.getHost(), host.getPort()));
		}

		if (gearmanClient.getSetOfJobServers().isEmpty()) {
			throw new GearmanUnreachableException("无法连接到Gearman的Job服务器");
		}

		isClentInited = true;

	}

	public GearmanClient getGearmanClient() {
		return gearmanClient;
	}

	@Override
	protected void finalize() throws Throwable {
		destroy();
		super.finalize();
	}
}
